package team.bluepen.supermarket.data.dao.abstraction;

import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;
import team.bluepen.supermarket.constant.Error;
import team.bluepen.supermarket.data.dao.DeleteRunner;
import team.bluepen.supermarket.data.entity.HbaseItem;
import team.bluepen.supermarket.data.entity.MarketTable;
import team.bluepen.supermarket.data.entity.Qualify;
import team.bluepen.supermarket.data.hbase.HTableRunner;
import team.bluepen.supermarket.data.hbase.HbaseConnectHelper;
import team.bluepen.supermarket.util.Log;
import team.bluepen.supermarket.util.StringUtil;

import javax.annotation.Nonnull;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * 表DAL操作抽象类
 * @author Kai
 */
public abstract class MarketDatabase<T extends HbaseItem<T>> implements DatabaseInterface<T> {
    protected boolean closeState;

    protected final Admin admin;
    protected final HTable table;
    protected final TableName name;
    protected final DeleteOperate<T> deleteOperate;
    protected final MarketTable marketTable;
    private static final String TAG = "MarketDatabase";

    protected MarketDatabase(@Nonnull MarketTable marketTable) throws IOException {
        closeState = false;
        this.marketTable = marketTable;
        Connection connection = HbaseConnectHelper.create();
        admin = connection.getAdmin();
        table = (HTable) connection.getTable(TableName.valueOf(marketTable.getName()));
        name = table.getName();
        deleteOperate = new DeleteRunner<>(admin, table);
    }

    public MarketTable getTable() {
        return marketTable;
    }

    protected abstract Put buildPut(T t);

    protected Error getTableState() {
        try {
            if (!admin.tableExists(name)){
                return Error.ERROR_TABLE_NOT_EXIST;
            }
            if (admin.isTableDisabled(name)) {
                return Error.ERROR_TABLE_DISABLED;
            }
        } catch (IOException e) {
            Log.e(TAG, e);
            return Error.ERROR_DATABASE;
        }
        return Error.SUCCESS;
    }

    /**
     * 附带建表操作。<br>
     * 如果需要对表进行其他操作，请使用 {@link HTableRunner}
     */
    public Error create() {
        try (HTableRunner hTableRunner = new HTableRunner(marketTable)) {
            return hTableRunner.create();
        } catch (IOException e) {
            Log.e(TAG, e);
            return Error.ERROR_DATABASE;
        }
    }

    @Override
    public Error insert(String family, String row, String qualifier, String value) {
        Error error = getTableState();
        if (!error.getState()) {
            return error;
        }

        if (StringUtil.isEmpty(family) ||
                StringUtil.isEmpty(row) ||
                StringUtil.isEmpty(qualifier) ||
                value == null){
            return Error.ERROR_NULL;
        }
        Put put = new Put(Bytes.toBytes(row));
        put.addColumn(Bytes.toBytes(family),
                Bytes.toBytes(qualifier),
                Bytes.toBytes(value));
        try {
            table.put(put);
            return Error.SUCCESS;
        } catch (IOException e) {
            Log.e(TAG, e);
            return Error.ERROR_DATABASE;
        }
    }

    @Override
    public Error insert(T t) {
        Error error = getTableState();
        if (!error.getState()) {
            return error;
        }

        if (t == null || StringUtil.isEmpty(t.getRow())){
            return Error.ERROR_NULL;
        }
        Put put = buildPut(t);
        try {
            table.put(put);
            return Error.SUCCESS;
        } catch (IOException e) {
            Log.e(TAG, e);
            return Error.ERROR_DATABASE;
        }
    }

    @Override
    public Error insert(List<T> ts) {
        if (ts == null){
            return Error.ERROR_NULL;
        }
        for (T _t: ts) {
            Error error = insert(_t);
            if (!error.getState()){
                return error;
            }
        }
        return Error.SUCCESS;
    }

    @Override
    public Error delete() {
        return deleteOperate.delete();
    }

    @Override
    public Error delete(String row) {
        return deleteOperate.delete(row);
    }

    @Override
    public Error delete(T t) {
        return delete(t.getRow());
    }

    @Override
    public Error delete(List<T> t) {
        return deleteOperate.delete(t);
    }

    @Override
    public Error delete(String row, String family) {
        return deleteOperate.delete(row, family);
    }

    protected abstract T getInformation(Result result);

    @Override
    public List<T> query() throws IOException {
        Error e = getTableState();
        if (!e.getState()) {
            Log.e(TAG, "An error occurred when querying. " + e);
            return new ArrayList<>();
        }
        List<T> ts = new ArrayList<>();
        Scan scan = new Scan();
        ResultScanner resultScanner = table.getScanner(scan);
        for (Result res : resultScanner){
            T t = getInformation(res);
            ts.add(t);
        }
        return ts;
    }

    @Override
    public T query(String row) throws IOException {
        Error e = getTableState();
        if (!e.getState()) {
            Log.e(TAG, "An error occurred when querying. " + e);
            return null;
        }
        if (StringUtil.isEmpty(row)) {
            return null;
        }

        Get get = new Get(Bytes.toBytes(row));
        Result result = table.get(get);
        return getInformation(result);
    }

    @Override
    public List<T> query(List<String> rows) throws IOException {
        if (rows == null) {
            return new ArrayList<>();
        }
        List<T> ts = new ArrayList<>();
        for (String row: rows) {
            ts.add(query(row));
        }
        return ts;
    }

    @Override
    public List<T> query(String family, String qualifier, byte[] value) throws IOException {
        if (StringUtil.isEmpty(family) || StringUtil.isEmpty(qualifier) ||
                value == null) {
            return new ArrayList<>();
        }
        Error e = getTableState();
        if (!e.getState()) {
            Log.e(TAG, "An error occurred when querying. " + e);
            return new ArrayList<>();
        }
        List<T> ts = new ArrayList<>();
        Scan scan = new Scan();
        scan.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier));
        SingleColumnValueFilter filter = new SingleColumnValueFilter(Bytes.toBytes(family)
                , Bytes.toBytes(qualifier), CompareFilter.CompareOp.EQUAL, value);
        filter.setFilterIfMissing(true);
        scan.setFilter(filter);
        ResultScanner scanner = table.getScanner(scan);
        for (Result result: scanner){
            T t = getInformation(result);
            t = query(t.getRow());
            ts.add(t);
        }
        return ts;
    }

    @Override
    public List<T> multiQuery(List<Qualify> qualifies) throws IOException {
        if (qualifies == null) {
            return new ArrayList<>();
        }

        Error e = getTableState();
        if (!e.getState()) {
            Log.e(TAG, "An error occurred when querying. " + e);
            return new ArrayList<>();
        }

        List<T> ts = new ArrayList<>();
        Scan scan = new Scan();
        FilterList filters = new FilterList();
        for (Qualify q: qualifies) {
            if (q == null || StringUtil.isEmpty(q.getFamily()) ||
                    StringUtil.isEmpty(q.getQualifier()) ||
                    q.getValue() == null || q.getBytesConverter() == null) {
                return new ArrayList<>();
            }
            Filter filter = new SingleColumnValueFilter(Bytes.toBytes(q.getFamily()),
                    Bytes.toBytes(q.getQualifier()), CompareFilter.CompareOp.EQUAL,
                    q.getBytesConverter().convert(q.getValue()));
            filters.addFilter(filter);
        }
        scan.setFilter(filters);
        ResultScanner scanner = table.getScanner(scan);
        for (Result result: scanner){
            T t = getInformation(result);
            t = query(t.getRow());
            ts.add(t);
        }
        return ts;
    }

    @Override
    public List<T> query(Qualify qualify) throws IOException {
        if (qualify == null) {
            return new ArrayList<>();
        }

        return query(qualify.getFamily(), qualify.getQualifier(),
                qualify.getBytesConverter().convert(qualify.getValue()));
    }

    @Override
    public void close() throws IOException {
        if (closeState) {
            return;
        }
        table.close();
        admin.close();
        closeState = true;
    }

    public boolean isOpen() {
        return !closeState;
    }

}
